package com.bkhech.bigdata.tiktok_anchor.videoinfotop10;

import cn.hutool.core.map.MapUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * 自定义实现Reducer类，在这里实现结果的聚合操作，取出直播时间最长的前 10 名主播以及直播时长
 *
 * @author guowm
 * @date 2022/7/6
 */
@Slf4j
public class VideoInfoTop10Reduce extends Reducer<Text, LongWritable, Text, LongWritable> {

    /**
     * 保存主播的ID以及器对应的开播总时长
     */
    Map<String, Long> map = new HashMap<>(10);

    /**
     * @param k2      主播ID
     * @param v2s     主播核心字段
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text k2, Iterable<LongWritable> v2s, Reducer<Text, LongWritable, Text, LongWritable>.Context context)
            throws IOException, InterruptedException {
        long lengthSum = 0;
        for (LongWritable v2 : v2s) {
            lengthSum += v2.get();
        }
        map.putIfAbsent(k2.toString(), lengthSum);
    }

    /**
     * 任务初始化的时候执行一次，仅执行一次，一般在里面做一些初始化资源链接的动作
     *
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Reducer<Text, LongWritable, Text, LongWritable>.Context context)
            throws IOException, InterruptedException {
        super.setup(context);
        log.info("execute once");
    }

    /**
     * 任务结束的时候执行一次，仅执行一次，做一些关闭资源的操作
     *
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void cleanup(Reducer<Text, LongWritable, Text, LongWritable>.Context context)
            throws IOException, InterruptedException {
        final Configuration conf = context.getConfiguration();
        // 从配置类中获取dt参数
        final String dt = conf.get(Cons.DT);
        // 根据map中的value进行排序
        Map<String, Long> sortByValueMap = MapUtil.sortByValue(map, true);
        sortByValueMap.entrySet().stream().limit(10).forEach(entry -> {
            String uid = entry.getKey();
            Long length = entry.getValue();
            // 封装 k3, v3
            Text k3 = new Text(dt + "\t" + uid);
            LongWritable v3 = new LongWritable(length);

            // 将聚合结果写出到 hdfs
            try {
                context.write(k3, v3);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    }
}
